package com.kafka.danYiMessage.yibu;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.SuccessCallback;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ExecutionException;

/**
 * 作者: LCG
 * 日期: 2020/6/19 18:09
 * 描述: 消息发送的确认回调
 */

@Component
public class KafKaYIbuProvider {

    @Autowired
   private KafkaTemplate<String,String> kafkaTemplate;

    /**
     * 消息发送回调 确保消息的发送
     * @param msg
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public void send(String msg) throws ExecutionException, InterruptedException {
        ProducerRecord<String,String> producerRecord=new ProducerRecord<String,String>("TOPIC",null,msg);

        //消息发送成功回调和失败回调 方式1（异步）
        kafkaTemplate.send(producerRecord).addCallback(new SuccessCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(SendResult<String, String> stringStringSendResult) {
                SimpleDateFormat sdf=new SimpleDateFormat("yyyy-mm-dd HH:mm:ss");
                String format = sdf.format(new Date(stringStringSendResult.getRecordMetadata().timestamp()));
                System.out.println("=====ok=====offset="+stringStringSendResult.getRecordMetadata().offset()+"==data="+format);
            }
        }, new FailureCallback() {
            @Override
            public void onFailure(Throwable throwable) {
                System.out.println("========onFailure=======");
            }
        });

       /* //消息发送成功回调和失败回调 方式2（异步）
        kafkaTemplate.send(producerRecord).addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                System.out.println("3========onFailure=======");
            }

            @Override
            public void onSuccess(SendResult<String, String> stringStringSendResult) {
                SimpleDateFormat sdf=new SimpleDateFormat("yyyy-mm-dd HH:mm:ss");
                String format = sdf.format(new Date(stringStringSendResult.getRecordMetadata().timestamp()));
                System.out.println("3=====ok=====offset="+stringStringSendResult.getRecordMetadata().offset()+"==data="+format);
            }
        });*/


    }

}
